package com.exmaple.cdc.factory;

import com.exmaple.cdc.annotation.CDCListener;
import com.exmaple.cdc.constant.CDCConstant;
import com.exmaple.cdc.listener.CDCMessageListener;
import com.exmaple.cdc.lock.MySqlLock;
import com.exmaple.cdc.service.SysColService;
import com.exmaple.cdc.service.SysMQService;
import com.exmaple.cdc.wrap.BinlogAddObj;
import com.exmaple.cdc.wrap.BinlogDeleteObj;
import com.exmaple.cdc.wrap.BinlogObj;
import com.exmaple.cdc.wrap.BinlogUpdateObj;
import com.exmaple.utils.TimeUtils;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.transaction.jta.JtaTransactionManager;

import javax.sql.DataSource;
import javax.transaction.*;
import java.io.Serializable;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Logger;

/*
 * @className: CDCServerFactory
 * @author: chentao
 * @date: 2025/3/7 下午2:34
 * @Version: 1.0
 * @description:
 */
public class CDCServerFactory implements BeanPostProcessor, InitializingBean, DisposableBean, BinaryLogClient.EventListener, BinaryLogClient.LifecycleListener {

    Logger logger = Logger.getLogger(getClass().getName());

    private Map<Long, String> tableMap = new ConcurrentHashMap<>();

    private Map<String, Object> listenerMap = new ConcurrentHashMap<>();

    private String host;

    private int port;

    private String db;

    private String username;

    private String password;

    private int serverId;

    private DataSource dataSource;

    private BinaryLogClient binaryLogClient;

    private ExecutorService executorService;

    private SysColService sysColService;

    private SysMQService sysMQService;

    private JtaTransactionManager jtaTransactionManager;

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        if (bean.getClass().isAnnotationPresent(CDCListener.class) && bean instanceof CDCMessageListener) {
            CDCListener dbListener = bean.getClass().getAnnotation(CDCListener.class);
            this.listenerMap.put(dbListener.tableName(), bean);
        }
        return BeanPostProcessor.super.postProcessAfterInitialization(bean, beanName);
    }

    /****
     * 初始化并启动
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        this.executorService = Executors.newCachedThreadPool();
        this.binaryLogClient = new BinaryLogClient(this.host, this.port, this.db, this.username, this.password);
        this.binaryLogClient.setServerId(serverId);
        this.binaryLogClient.registerEventListener(this);
        this.binaryLogClient.registerLifecycleListener(this);
        this.executorService.execute(new ServerStartThread());
    }

    /****
     * 删除并停止链接
     * @throws Exception
     */
    @Override
    public void destroy() throws Exception {
        this.binaryLogClient.disconnect();
        if (!this.executorService.isShutdown()) {
            this.executorService.shutdownNow();
        }
    }

    /***
     * 数据库表变化的消息监听
     * @param event
     */
    @Override
    public void onEvent(Event event) {
        if (event.getData() instanceof TableMapEventData) {
            TableMapEventData eventData = (TableMapEventData) event.getData();
            long tableId = eventData.getTableId();
            String tableName = eventData.getTable();
            this.tableMap.putIfAbsent(tableId, tableName);
        } else {
            BinlogObj binlogObj = this.wrapBinlogObj(event);
            if (binlogObj != null && this.listenerMap.containsKey(binlogObj.getTableName())) {
                this.executorService.execute(new CDCHandlerThread(binlogObj));
            }
        }
    }

    /***
     * 封裝日志結果
     * @return
     */
    private BinlogObj wrapBinlogObj(Event event) {
        BinlogObj binlogObj = null;
        //数据的更新
        if (event.getData() instanceof UpdateRowsEventData) {
            UpdateRowsEventData updateRowsEventData = event.getData();
            String tableName = this.tableMap.get(updateRowsEventData.getTableId());
            binlogObj = new BinlogUpdateObj(tableName, CDCConstant.UPDATE, updateRowsEventData);
            if (this.listenerMap.containsKey(tableName)) {
                Map<Long, String> cols = sysColService.getColsMap(db, tableName);
                for (Map.Entry<Serializable[], Serializable[]> row : updateRowsEventData.getRows()) {
                    Map<String, Map<Serializable, Serializable>> colMap = new HashMap<>();
                    Serializable[] key = row.getKey();
                    Serializable[] value = row.getValue();
                    for (int i = 0; i < key.length; i++) {
                        Map<Serializable, Serializable> valMap = new HashMap<>();
                        String colName = cols.get((long) i);
                        valMap.put(key[i], value[i]);
                        colMap.put(colName, valMap);
                    }
                    ((BinlogUpdateObj) binlogObj).getUpdateRowList().add(colMap);
                }
            }
        }
        //数据的新增
        if (event.getData() instanceof WriteRowsEventData) {
            WriteRowsEventData writeRowsEventData = event.getData();
            String tableName = this.tableMap.get(writeRowsEventData.getTableId());
            binlogObj = new BinlogAddObj(tableName, CDCConstant.ADD, event.getData());
            if (this.listenerMap.containsKey(tableName)) {
                Map<Long, String> cols = sysColService.getColsMap(db, tableName);
                BitSet includedColumns = writeRowsEventData.getIncludedColumns();
                for (Serializable[] row : writeRowsEventData.getRows()) {
                    Map<String, Serializable> map = new HashMap<>();
                    for (int i = 0; i < includedColumns.length(); i++) {
                        int colIndex = includedColumns.nextSetBit(i);
                        map.put(cols.get((long) colIndex), row[colIndex]);
                    }
                    ((BinlogAddObj) binlogObj).getRowsList().add(map);
                }
            }
        }
        //数据的删除
        if (event.getData() instanceof DeleteRowsEventData) {
            DeleteRowsEventData deleteRowsEventData = event.getData();
            String tableName = this.tableMap.get(deleteRowsEventData.getTableId());
            binlogObj = new BinlogDeleteObj(tableName, CDCConstant.DELETE, event.getData());
            if (this.listenerMap.containsKey(tableName)) {
                Map<Long, String> cols = sysColService.getColsMap(db, tableName);
                BitSet includedColumns = deleteRowsEventData.getIncludedColumns();
                List<Serializable[]> rows = deleteRowsEventData.getRows();
                for (Serializable[] row : rows) {
                    Map<String, Serializable> map = new HashMap<>();
                    for (int i = 0; i < includedColumns.length(); i++) {
                        int colIndex = includedColumns.nextSetBit(i);
                        map.put(cols.get((long) colIndex), row[colIndex]);
                    }
                    ((BinlogDeleteObj) binlogObj).getRowsList().add(map);
                }
            }
        }
        return binlogObj;
    }

    /***
     * 有客户端连接上消息监听
     * @param client the client that logged in
     */
    @Override
    public void onConnect(BinaryLogClient client) {
        logger.warning("server : " + client.getServerId() + " connect success ...");
    }

    /***
     * 客户端连接失败消息监听
     * @param client the client that triggered the communication failure
     * @param ex The exception that triggered the communication failutre
     */
    @Override
    public void onCommunicationFailure(BinaryLogClient client, Exception ex) {
        logger.warning("server : " + client.getServerId() + " connect failure ...");
    }

    /****
     * 序列化失败的监听
     * @param client the client that failed event deserialization
     * @param ex The exception that triggered the failutre
     */
    @Override
    public void onEventDeserializationFailure(BinaryLogClient client, Exception ex) {
        //序列化失败
    }

    /***
     * 连接关闭的消息监听
     * @param client the client that disconnected
     */
    @Override
    public void onDisconnect(BinaryLogClient client) {
        logger.warning("server : " + client.getServerId() + " disconnect ...");
    }

    /****
     * 服务启动线程
     */
    class ServerStartThread implements Runnable {
        @Override
        public void run() {
            try {
                binaryLogClient.connect();
            } catch (Exception e) {
                logger.warning("binaryLogClient 客户端启动失败: " + e.getMessage());
            }
        }
    }

    /***
     * 消息监听处理类
     */
    class CDCHandlerThread implements Runnable {

        private BinlogObj binlogObj;

        public CDCHandlerThread(BinlogObj binlogObj) {
            this.binlogObj = binlogObj;
        }

        @Override
        public void run() {
            String tableName = binlogObj.getTableName();
            Object bean = listenerMap.get(tableName);
            if (binlogObj instanceof BinlogAddObj && bean.getClass().isAnnotationPresent(CDCListener.class) && bean instanceof CDCMessageListener) {
                CDCListener cdcListener = bean.getClass().getAnnotation(CDCListener.class);
                CDCMessageListener cdcMessageListener = (CDCMessageListener) bean;
                if (cdcListener.multiple()) {
                    handle(binlogObj, cdcMessageListener);
                } else {
                    MySqlLock lock = null;
                    try {
                        lock = new MySqlLock(dataSource);
                        lock.lock();
                        if (!sysMQService.isCommit(binlogObj)) {
                            handle(binlogObj, cdcMessageListener);
                        }
                    } catch (Exception e) {
                        logger.warning("mq消息消费失败: " + e.getMessage());
                    } finally {
                        if (lock != null) {
                            lock.unlock();
                        }
                    }
                }
            }
        }
    }

    //消息重试线程
    class CDCRetryHandleThread implements Runnable {

        private Map<String, Object> map;

        public CDCRetryHandleThread(Map<String, Object> map) {
            this.map = map;
        }

        @Override
        public void run() {

        }
    }

    private void handle(BinlogObj binlogObj, CDCMessageListener cdcMessageListener) {
        UserTransaction userTransaction = null;
        try {
            userTransaction = jtaTransactionManager.getUserTransaction();
            userTransaction.begin();
            cdcMessageListener.onMessage(binlogObj);
            if (!sysMQService.isCommit(binlogObj)) {
                sysMQService.commitObj(binlogObj);
            }
        } catch (Exception e) {
            sysMQService.rollBackObj(binlogObj);
            logger.warning("mq消息消费失败: " + e.getMessage());
            try {
                userTransaction.rollback();
                userTransaction = null;
            } catch (Exception ex) {
                logger.warning("mq消息消费失败: " + e.getMessage());
            }
        } finally {
            if (userTransaction != null) {
                try {
                    userTransaction.commit();
                } catch (Exception e) {
                    logger.warning("mq消息消费失败: " + e.getMessage());
                }
            }
        }
    }

    /***
     * 消息重试
     * @param table
     * @param map
     * @param cdcMessageListener
     */
    private void handle(String table, Map<String, Object> map, CDCMessageListener cdcMessageListener) {
        Long id = (Long) map.get(CDCMessageListener.ID);
        int retryCount = (Integer) map.get(CDCMessageListener.RETRY_COUNT);
        UserTransaction userTransaction = null;
        try {
            userTransaction = jtaTransactionManager.getUserTransaction();
            userTransaction.begin();
            cdcMessageListener.failHandler(map);
            if (!sysMQService.isCommit(table, id)) {
                sysMQService.commit(table, id);
            }
        } catch (Exception e) {
            logger.warning("mq消息消费失败: " + e.getMessage());
            retryCount = retryCount + 1;
            Date retryTime = TimeUtils.getTime(retryCount * 60);
            try {
                userTransaction.rollback();
                userTransaction = null;
            } catch (Exception ex) {
                logger.warning("mq消息消费失败: " + e.getMessage());
            }
            sysMQService.rollBack(table, id, retryCount, retryTime);
        } finally {
            if (userTransaction != null) {
                try {
                    userTransaction.commit();
                } catch (Exception e) {
                    logger.warning("mq消息消费失败: " + e.getMessage());
                }
            }
        }
    }

    //启动重试任务
    private void retryRunLog() {
        for (String tableName : this.listenerMap.keySet()) {
            Object bean = this.listenerMap.get(tableName);
            if (bean.getClass().isAnnotationPresent(CDCListener.class) && bean instanceof CDCMessageListener) {
                CDCListener cdcListener = bean.getClass().getAnnotation(CDCListener.class);
                CDCMessageListener cdcMessageListener = (CDCMessageListener) bean;
                List<Map<String, Object>> mapList = sysMQService.queryList(cdcListener.tableName(), CDCMessageListener.INIT);
                for (Map<String, Object> map : mapList) {
                    handle(tableName, map, cdcMessageListener);
                }
            }
        }
    }

    // set get

    public String getHost() {
        return host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public String getDb() {
        return db;
    }

    public void setDb(String db) {
        this.db = db;
    }

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public SysColService getSysColService() {
        return sysColService;
    }

    public void setSysColService(SysColService sysColService) {
        this.sysColService = sysColService;
    }

    public int getServerId() {
        return serverId;
    }

    public void setServerId(int serverId) {
        this.serverId = serverId;
    }

    public DataSource getDataSource() {
        return dataSource;
    }

    public void setDataSource(DataSource dataSource) {
        this.dataSource = dataSource;
    }

    public SysMQService getSysMQService() {
        return sysMQService;
    }

    public void setSysMQService(SysMQService sysMQService) {
        this.sysMQService = sysMQService;
    }

    public JtaTransactionManager getJtaTransactionManager() {
        return jtaTransactionManager;
    }

    public void setJtaTransactionManager(JtaTransactionManager jtaTransactionManager) {
        this.jtaTransactionManager = jtaTransactionManager;
    }
}
